/* 
   异步工作器
   */
#ifndef __CLX_LOOPER_H__
#define __CLX_LOOPER_H__

#include "util.hpp"
#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <memory>
#include <atomic>
#include <functional>
#include <condition_variable>

namespace clxlog{
  using Functor = std::function<void(Buffer &)>;
  enum class AsynchType{ ASYNC_SAFE, ASYNC_UNSAFE }; // 安全状态（缓冲区满了就阻塞）/非安全状态（缓冲区满了就扩容，不考虑资源耗尽的风险）
  class AsynchLooper {
    public:
      using ptr = std::shared_ptr<AsynchLooper>;
      AsynchLooper(const Functor &cb, AsynchType looper_type = AsynchType::ASYNC_SAFE) 
        :_stop(false), _thread(std::thread(&AsynchLooper::threadEntry, this)), _callBack(cb), _looper_type(looper_type) {}
      ~AsynchLooper(){ stop(); }
      void stop() { 
        _stop = true;                    // 退出标志设置为true 
        _consumer_cond.notify_all();     // 唤醒所有工作线程
        _thread.join();                  // 等待工作线程退出
      }
      void push(const char *data, size_t len) {
        // 1、无限扩容 非安全(资源耗尽，程序崩溃) 2、固定大小 - 生产缓冲区中数据满了就阻塞
        std::unique_lock<std::mutex> lock(_mutex);
        /* 安全状态 */
        if (_looper_type == AsynchType::ASYNC_SAFE) {
          _produce_cond.wait(lock, [&](){ return _produce_buffer.writeAbleSize() >= len; });
        }
        _produce_buffer.push(data, len);
        _consumer_cond.notify_one();   //  唤醒消费者对缓冲区中的数据进行处理
      }
    private:
      /* 线程的入口函数 -- 对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区*/
      void threadEntry() {
        while (1) {
          {
            // 1、 判断生产缓冲区有没有数据，有则交换，无则阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            /* 退出标志被设置了，且缓冲区没有数据，这时候退出，否则可能缓冲区中任然有数据未被完全处理 */
            if (_stop && _produce_buffer.bufferEmpty()) break;
            _consumer_cond.wait(lock, [&](){ return (_stop || !_produce_buffer.bufferEmpty()); });
            _produce_buffer.bufferSwap(_consumer_buffer);
            // 2、 唤醒生产者(只有安全状态生产者才会被阻塞)
            if (_looper_type == AsynchType::ASYNC_SAFE) {
              _produce_cond.notify_all();
            }
          }
          // 3、 被唤醒后，对消费缓冲区进行数据处理(处理过程无需加锁保护)
          _callBack(_consumer_buffer);
          // 4、 初始化消费缓冲区
          _consumer_buffer.bufferReset();
        }
      }  
      /* 回调函数 具体对缓冲区数据进行处理的回调函数， 由异步工作器的使用者传入 */
      Functor _callBack; 

    private:
      AsynchType _looper_type;
      std::atomic<bool> _stop;                 // 工作器停止标志
      std::mutex _mutex;          
      Buffer _produce_buffer;
      Buffer _consumer_buffer;
      std::condition_variable _produce_cond;
      std::condition_variable _consumer_cond;
      std::thread _thread;        
  };
}

#endif
